package com.shujia.sql

import org.apache.spark.sql.catalyst.expressions.CaseWhen
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo3DFAPI13 {
  def main(args: Array[String]): Unit = {
    // 构建SparkSession
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo3DFAPI13")
      .master("local")
      .config("spark.sql.shuffle.partitions", 2) // 如果不指定默认shuffle会有200个并行度
      .getOrCreate()

    val stuDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id Int,name String,age Int,gender String,clazz String")
      .load("Spark/data/stu/students.txt")

    // stuDF被多次使用 可以进行Cache
    stuDF.cache()

    // 导入隐式转换
    import spark.implicits._
    // 导入Spark SQL默认提供的函数
    import org.apache.spark.sql.functions._

    // DSL
    // 过滤出年龄>23的学生
    stuDF
      .where("age > 23") // 字符串表达式
    //      .show(10)

    stuDF
      .where($"age" > 23) // 列表达式
    //      .show(10)

    // 过滤出文科班的学生
    stuDF
      .where($"clazz" like "文科%") // 列表达式
    //      .show()

    stuDF
      .filter(row => {
        val age: Int = row.getAs[Int]("age")
        age > 23
      })
    //      .show()

    // select
    stuDF
      .select($"id" as "stuId", when($"age" > 18, "成年").when($"age" < 18, "未成年") as "是否成年")
    //      .show(10)

    // 分组聚合
    // groupBy会产生Shuffle，在Spark SQL中默认会使用200个任务对Shuffle任务进行处理
    stuDF
      .groupBy($"clazz")
      .count()
    //      .show()

    stuDF
      .groupBy($"clazz".substr(1, 2) as "clazz")
      .agg(countDistinct($"id") as "cnt")
    //      .show()

    stuDF
      .select($"clazz".substr(1, 2) as "clazz", expr("1") as "c")
      .groupBy($"clazz")
      .agg(sum($"c") as "cnt")
    //      .show()

    // join
    val scoDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id Int,sid Int,score Int")
      .load("Spark/data/stu/score.txt")

    // 如果关联的列名一致 则可以直接取
    //    stuDF
    //      .join(scoDF, $"id" === $"stuId", "left")
    //      .show()

    // 如果列名一致 并且需要指定关联类型 则可以将 重名的列通过List传入
    stuDF
      .join(scoDF, List("id"), "left")
      .show()



    // 用完记得释放
    stuDF.unpersist()
  }

}
